package com.han.elasticsearch.kafka;

import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.han.elasticsearch.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author hanjl
 * @Description Kafka生产者
 * @ClassName KafkaProducer
 * @createTime 2021/12/22 16:39
 */
@Component
@Slf4j
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String,String> kafkaTemplate;

    /**
     * 批量发送数据
     * @param messages
     * @param topic
     * @param <T>
     */
    public <T> void sendMessage(List<T> messages,String topic){
        if(CollectionUtils.isEmpty(messages)){
            log.error("topic {},message为空",topic);
            return;
        }
        if(StringUtils.isBlank(topic)){
            log.error("topic不可以为空");
        }
        log.info("准备向topic：{} 发送 {} 条数据",topic,messages.size());
        kafkaTemplate.send(topic, JsonUtil.toJsonString(messages));
        log.info("向topic：{} 发送 {} 条数据成功",topic,messages.size());
    }

    /**
     * 带回调的生产者1
     * @param messages
     * @param topic
     * @param <T>
     */
    public <T> void sendMessageCallBack(List<T> messages,String topic){
        if(CollectionUtils.isEmpty(messages)){
            log.error("topic {},message为空",topic);
            return;
        }
        if(StringUtils.isBlank(topic)){
            log.error("topic不可以为空");
        }
        log.info("准备向topic：{} 发送 {} 条数据",topic,messages.size());

        kafkaTemplate.send(topic, JsonUtil.toJsonString(messages)).addCallback(success -> {
            // 消息发送到的topic
            String topic1 = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("发送消息成功:" + topic1 + "-" + partition + "-" + offset);
        },failure -> {
            System.out.println("发送消息失败：" + failure.getMessage());
        });
        log.info("向topic：{} 发送 {} 条数据成功",topic,messages.size());
    }

    /**
     * 带回调的生产者2
     * @param messages
     * @param topic
     * @param <T>
     */
    public <T> void sendMessageCallBack2(List<T> messages,String topic){
        if(CollectionUtils.isEmpty(messages)){
            log.error("topic {},message为空",topic);
            return;
        }
        if(StringUtils.isBlank(topic)){
            log.error("topic不可以为空");
        }
        log.info("准备向topic：{} 发送 {} 条数据",topic,messages.size());

        kafkaTemplate.send(topic, JsonUtil.toJsonString(messages)).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败：" + ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                        + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
            }
        });
        log.info("向topic：{} 发送 {} 条数据成功",topic,messages.size());
    }

    public <T> void sendMessageString(String messages,String topic){
        if(StringUtils.isBlank(messages)){
            log.error("topic {},message为空",topic);
            return;
        }
        if(StringUtils.isBlank(topic)){
            log.error("topic不可以为空");
        }
        log.info("准备向topic：{} 发送 {} 数据",topic,messages);
        kafkaTemplate.send(topic, JsonUtil.toJsonString(messages));
        log.info("向topic：{} 发送 {} 数据成功",topic,messages);
    }
}
